package net.bwie.realtime.jtp.dwd.order.utils;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.HashMap;
import java.util.Map;
/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/06/03 21:28:57
 * @version: 1.0
 */
public class HbaseUtil {
    private static Connection connection = null ;

    static {
        try{
            // 1.创建配置对象，设置属性值，比如连接ZK集群信息
            org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create() ;
            conf.set(HConstants.CLIENT_ZOOKEEPER_QUORUM, "node101,node102,node103");
            conf.set(HConstants.CLIENT_ZOOKEEPER_CLIENT_PORT, "2181");
            // 2.实例化Connection连接
            connection = ConnectionFactory.createConnection(conf) ;
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 依据RowKey获取HBase表中数据
     * @param tableName 表名称
     * @param rowKey 主键
     * @param familyName 列簇名称
     * @param columnName 列名称
     */
    public static String getData(String tableName, String rowKey,
                                 String familyName, String columnName) throws Exception{
        // a.实例化Table对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        // b. 获取数据
        Get get = new Get(Bytes.toBytes(rowKey));
        Result result = table.get(get);
        // c. 获取字段值
        byte[] resultValue = result.getValue(Bytes.toBytes(familyName), Bytes.toBytes(columnName));
        table.close();
        // d. 返回
        return Bytes.toString(resultValue) ;
    }

    /**
     * 扫描HBase表数据
     * @param tableName 表名称
     * @param family 列簇名称
     */
    public static Map<String, String> scanData(String tableName, String family, String column) throws Exception{
        // a.实例化Table对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        // b.构建扫描器对象
        Scan scan = new Scan() ;
        scan.addFamily(Bytes.toBytes(family)) ;
        // c. 扫描数据
        ResultScanner scanner = table.getScanner(scan);
        // d. 提取数据
        Map<String, String> dimMap = new HashMap<>() ;
        for (Result result : scanner) {
            // RowKey行健
            String rowKey = Bytes.toString(result.getRow());
            // 列值
            String columnValue = Bytes.toString(result.getValue(Bytes.toBytes(family), Bytes.toBytes(column)));
            // 放入集合
            dimMap.put(rowKey, columnValue);
        }
        // e. 关闭
        scanner.close();
        table.close();
        // f. 返回对象
        return dimMap ;
    }


    /**
     * 关闭链接
     */
    public static void close() throws Exception {
        if(null != connection){
            connection.close();
        }
    }
}
